package com.youxin.dataStream.custormSource;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * 丰富的source 其实就是多了open 和 close
 */
public class RichParalleSource extends RichSourceFunction<Long> {

    private boolean flag = true;
    private Long num = 0L;



    /**
     * 持续输出数据的方法
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceFunction.SourceContext<Long> sourceContext) throws Exception {
        while(flag){
            sourceContext.collect(num++);
            Thread.sleep(1000);
        }
    }

    /**
     * 关闭程序的时候调用一次
     */
    @Override
    public void cancel() {
        flag = false;
    }

    /**
     * 资源初始化
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        System.out.println("open....");
    }

    /**
     * 关闭资源
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        super.close();
        System.out.println("close....");
    }
}
